import json
import asyncio
from typing import Optional

import aio_pika
from aio_pika import Message
import pickle

from uvicorn.middleware.message_logger import message_with_placeholders

from core.dependencies import get_redis_service, get_sender_tools, get_minio_client, get_rabbitmq_channel
from core.logger import log
from scheduler.msg_async_sch import MsgAsyncScheduler
from services.wx_chat_messages import create_wx_messages, get_wx_messages_by_session_id
from services.wx_friends_service import get_fd_by_wx_id, get_friend_nickname, set_friends_remark
from utils.msg_filter_util import filter_msg_by_type,fiter_special_chat
from utils.ollama import generate_description
from utils.rag_api import get_reply_by_rag
from socketio import AsyncServer
from utils.socket_util import broadcast_message, download_image
from db.models.chat_message import MqMessage, AndroidAppMQMessage



class RabbitMQConsumerCallTask:
    def __init__(self,channel:str=None):
        self.consume_task = None
        self.channel = channel
        self.is_consuming = False
        self.consumer_tag = None
    async def callback(self,meassge:aio_pika.IncomingMessage):
        """处理轮呼消息队列消息"""
        async with meassge.process():  # 自动确认消息
            try:
                meassge_data = json.loads(meassge.body.decode('utf-8'))
                await self._process_call_task_meassge(meassge_data)
            except Exception as e:
                log.error(f"处理消息时出错: {str(e)}")
                log.info(meassge.body.decode('utf-8'))


    async def _process_call_task_meassge(self,message_data:dict):
        """具体处理轮呼任务的消息逻辑"""
        pass
    async def start_consuming(self,task_id:str=None):
        """开始消费消息"""
        if not self.channel:
            raise ValueError("必须指定channel")
        queue_name = f"call_task_{self.channel}"
        async with get_rabbitmq_channel() as channel:
            queue = await channel.declare_queue(queue_name, durable=True)

            #设置QoS
            await channel.set_qos(prefetch_count=1)

            #开始消费
            self.consumer_tag = await queue.consume(self.callback,no_ack= False)
            self.is_consuming = True
            log.info(f"开始消费消息队列：{queue_name}")
    async def stop_consuming(self):
        """停止消费消息"""
        self.is_consuming = False
        if self.consumer_tag:
            async with get_rabbitmq_channel() as channel:
                await channel.cancel(self.consumer_tag)
        log.info(f"停止消费消息队列")

    async def selective_clear_queue(self,fiter_func):
        """选择性删除队列中的消息"""
        if not self.channel:
            raise ValueError("必须指定channel")
        queue_name = f"call_task_{self.channel}"
        messages_to_keep = []
        async with get_rabbitmq_channel() as channel:
            # 先声明队列
            queue = await channel.declare_queue(queue_name, durable=True)
            while True:
                try:
                    message = await queue.get(no_ack=False, fail=False)
                    if not message:
                        break
                    # method,properties,body = message
                    try:
                        message_data = json.loads(message.body.decode('utf-8'))
                        if fiter_func(message_data):
                            await message.ack()
                        else:
                            messages_to_keep.append((message.body, message.properties))
                            await message.nack(requeue=False)
                    except Exception as e:
                        # 出错时重新入队
                        await message.nack(requeue=True)
                except Exception as e:
                    log.error(f"读取消息失败: {e}")
                    break
            # 重新发送保留的消息
            for body, properties in messages_to_keep:
                m = Message(
                    body=body,
                    delivery_mode=aio_pika.DeliveryMode.PERSISTENT,
                )
                await channel.default_exchange.publish(m, routing_key=queue.name)

class RabbitMQConsumer:
    def __init__(self, topic_name: str, scheduler:MsgAsyncScheduler, sio_service:AsyncServer):
        self.topic_name = topic_name
        self.queue_name = f"{topic_name}_wx_sync_msg_topic"
        self.consumer_tag = None
        self.is_consuming = False
        self.channel = None
        self.server = None
        self.sender_tools = None
        self.minio_client = None
        self.consume_task = None  # 添加任务引用
        self.scheduler = scheduler
        self.sio = sio_service  # 通过构造函数传入
    async def init(self):
        """异步初始化依赖服务"""
        self.server = await get_redis_service()
        self.sender_tools = get_sender_tools()
        self.minio_client = get_minio_client()
    async def callback(self, message: aio_pika.IncomingMessage):
        """
        异步消息处理回调函数
        Args:
            message: 接收到的消息
        """
        async with (message.process()):  # 自动确认消息
            try:
                # 对消息进行预处理--去重，过滤不是文本，图片消息
                AddMsgs = await self._first_process_msg(message)
                for msg in AddMsgs:
                    # 判断消息发送者是否是当前账号
                    if not msg.current_user==msg.from_user:
                        #不是本人自己发的消息
                        log.info(f"消息发送者：{msg.from_user}，消息接收者：{msg.to_user}，消息内容：{msg.content}，消息ID：{msg.msg_id}")

                        # 准备要广播的消息数据
                        sender_nickname = await get_friend_nickname(friend_wx_id=msg.from_user,device_id=msg.to_user)
                        # 广播消息到 WebSocket 设备房间
                        broadcast_data = {
                            "device_id": msg.to_user,
                            "wx_id": msg.from_user,
                            "device_nickname": msg.current_user,
                            "sender_nickname": sender_nickname if sender_nickname else msg.from_user,
                            "content": msg.content,
                            "type": msg.msg_type,
                            "timestamp": msg.create_time,
                            "msg_id": msg.msg_id
                        }
                        #将新消息通知到当前设备下的所有用户
                        await broadcast_message(self.sio,broadcast_data,room=msg.to_user)

                        # 获取当前发消息人的详情信息（从数据库中查找）
                        wx_fr = await get_fd_by_wx_id(msg.from_user, msg.to_user)

                        # 判断当前发消息的人是否是客户
                        if "客户" in wx_fr.tags:
                            await self._process_custom_msg(msg)
                        else:
                            await self._process_general_msg(msg)
                    else:
                        await self._process_self_msg(msg)
            except ValueError as  e:
                log.info(f"消息处理错误：{e}")
            except Exception as e:
                log.error(f"处理消息时出错: {str(e)}")
                log.info(message.body.decode('utf-8'))

    async def _consume_loop(self):
        """消费循环"""
        async with get_rabbitmq_channel() as channel:
            # 声明队列
            queue = await channel.declare_queue(
                self.queue_name,
                durable=True
            )
            # 设置QoS
            await channel.set_qos(prefetch_count=1)

            # 开始消费
            self.consumer_tag = await queue.consume(self.callback, no_ack=False)
            self.is_consuming = True

            log.info(f"开始消费队列: {self.queue_name}")

            # 保持任务持续运行
            while self.is_consuming:
                await asyncio.sleep(3)


    async def start_consuming(self):
        """启动消费者"""
        if self.consume_task is None or self.consume_task.done():
            self.consume_task = asyncio.create_task(self._consume_loop())
        return self.consume_task


    async def stop_consuming(self):
        """停止消费者"""
        self.is_consuming = False  # 设置停止标志

        if self.consume_task and not self.consume_task.done():
            # 直接取消任务，因为它是无限循环的
            self.consume_task.cancel()
            try:
                await self.consume_task  # 等待任务真正取消
            except asyncio.CancelledError:
                pass  # 这是预期的行为

        self.consume_task = None
        self.consumer_tag = None
        log.info(f"消费者已停止: {self.queue_name}")


    async def _first_process_msg(self, message)->list[MqMessage]:
        # 解析消息 消息预处理，将消息统一格式
        raw_data = json.loads(message.body.decode('utf-8'))
        processed_data = []
        if not raw_data.get("AddMsgs"):
            raise ValueError("消息格式错误")
        AddMsgs = raw_data["AddMsgs"]
        current_user = raw_data["userName"]
        key = raw_data["UUID"]
        for msg in AddMsgs:
            content = msg["content"]["str"]
            from_user = msg["from_user_name"]["str"]
            to_user = msg["to_user_name"]["str"]
            msg_id = msg["msg_id"]
            msg_type = msg["msg_type"]
            create_time = msg["create_time"]
            # 对消息类型进行过滤（如果不是文本和图片消息，则忽略）
            if await filter_msg_by_type(msg_type):
                continue

            # 对mq进行去重，防止重启导致重复消息
            if await self.server.mq_duplicate_msg(msg_id):
                log.info(f"重复消息，已忽略，消息内容：{content}，消息ID：{msg_id}")
                continue
            # 对消息进行去重
            if await self.server.is_duplicate_msg(from_user, to_user, content):
                continue
            # 将特定的一些消息进行提前处理--> 例如：遇到（你好）直接回复good,//或者当触发某些关键字，就直接不回复或者回复图片----不走模型
            if await fiter_special_chat(content,from_user,key):
                continue
            data = {
                "from_user": from_user,
                "to_user": to_user,
                "content": content,
                "msg_id": msg_id,
                "msg_type": msg_type,
                "create_time": create_time,
                "current_user": current_user,
                "key": key
            }
            mq_msg = MqMessage(**data)
            processed_data.append(mq_msg)
        return processed_data


    async def _process_custom_msg(self, msg:MqMessage):
        # 保存接受消息到数据库
        if msg.msg_type == 3:
            # 将图片上传保存到数据库（以url格式保存）
            image_url = await download_image(msg.key, msg.from_user, msg.msg_id, msg.to_user, self.minio_client)
            await create_wx_messages(msg.from_user, msg.to_user, "接收", image_url, content_type=msg.msg_type)
            await self.server.set_image_url(image_url, msg.from_user, msg.to_user)
            # 启动监控任务
            await self.scheduler.start_monitoring_task(msg.from_user, msg.to_user, data_type="image_url", key=msg.key)
            return
        elif msg.msg_type == 1:
            if f"{msg.from_user}:{msg.to_user}" in self.scheduler.monitoring_states:
                # 停止定时任务
                await self.scheduler.stop_monitoring_task(msg.from_user, msg.to_user)
                await self.scheduler.process_list_data(msg.from_user, msg.to_user, data_type="image_url", key=msg.key,next_content=msg.content)
                return
        # 保存接受消息到数据库
        await create_wx_messages(msg.from_user, msg.to_user, "接收", msg.content)
        # 开始回复
        reply = await get_reply_by_rag(msg.content, msg.from_user, msg.to_user)
        if "##" in reply:
            history_msgs = await get_wx_messages_by_session_id(sender_id=msg.from_user, receiver_id=msg.to_user)
            result = await generate_description(history_msgs)
            await set_friends_remark(msg.to_user, msg.from_user, result)
            return
        send = self.sender_tools.send_wx_message(reply, msg.from_user, msg.key)
        log.info(f"发送结果: {send}")
        await create_wx_messages(msg.to_user, msg.from_user, "回复", reply)
        send_broadcast_data = {
            "device_id": msg.to_user,
            "is_self_message": True,
            "content": reply,
            "wx_id": msg.from_user
        }
        await broadcast_message(self.sio, send_broadcast_data, room=msg.to_user)


    async def _process_general_msg(self, msg: MqMessage):
        # 不是自己发的，也不是客户标签的人发的
        if msg.msg_type == 1:
            await create_wx_messages(msg.from_user, msg.to_user, "接收", msg.content)
        elif msg.msg_type == 3:
            # 首先获取图片二进制数据
            image_url = await download_image(msg.key, msg.from_user, msg.msg_id, msg.to_user, self.minio_client)
            await create_wx_messages(msg.from_user, msg.to_user, "接收", image_url, content_type=msg.msg_type)


    async def _process_self_msg(self, msg: MqMessage):
        # 自己发的消息
        log.info(f"消息发送者：{msg.from_user}，消息接收者：{msg.to_user}，消息内容：{msg.content}，消息ID：{msg.msg_id}")
        if msg.msg_type == 1:
            broadcast_data = {
                "is_self_message": True,
                "wx_id": msg.to_user,
                "device_id": msg.from_user,
                "content": msg.content,
                "timestamp": msg.create_time,
                "type": msg.msg_type
            }
            await create_wx_messages(msg.from_user, msg.to_user, "回复", msg.content)
            # 广播消息到 WebSocket 设备房间
            await broadcast_message(self.sio, broadcast_data, room=msg.from_user)

    async def clear_mq_data(self):
        """清空MQ数据"""

        await self.server.clear_mq_data()



class RabbitMQProductor:

    async def productor_message(self, message: AndroidAppMQMessage):

        async with get_rabbitmq_channel() as channel:
            exchange = await channel.declare_exchange(
                name="AndroidMQ",
                type="direct",
                durable=True
            )

            queue = await channel.declare_queue(
                name="call_task_"+message.device_id,
                durable=True
            )

            await queue.bind(exchange=exchange, routing_key=queue.name)

            m = Message(
                body=message.model_dump_json().encode("utf-8"),
                delivery_mode=aio_pika.DeliveryMode.PERSISTENT,
            )
            await exchange.publish(m, routing_key=queue.name)

    async def productor_message_batch(self, messages: list[AndroidAppMQMessage]):
        """批量生产消息"""
        if not messages:
            return

        async with get_rabbitmq_channel() as channel:
            # 只声明一次交换机
            exchange = await channel.declare_exchange(
                name="AndroidMQ",
                type="direct",
                durable=True
            )

            # 使用字典缓存已声明的队列
            queue_cache = {}

            for message in messages:
                device_id = message.device_id
                queue_name = f"call_task_{device_id}"

                # 检查队列是否已缓存
                if queue_name not in queue_cache:
                    queue = await channel.declare_queue(
                        name=queue_name,
                        durable=True
                    )
                    await queue.bind(exchange=exchange, routing_key=queue_name)
                    queue_cache[queue_name] = queue

                # 发布消息
                try:
                    m = Message(
                        body=message.model_dump_json().encode("utf-8"),
                        delivery_mode=aio_pika.DeliveryMode.PERSISTENT,
                    )
                    await exchange.publish(m, routing_key=queue_name)
                except Exception as e:
                    log.error(f"发布消息失败: {e}")
                    raise  e


